/**
 * Package name:indi.yangshenhui.hbase.test
 * File name:RuJiaTask.java
 * Date:2016年4月9日-下午4:20:51
 * feiniu.com Inc.Copyright (c) 2013-2015 All Rights Reserved.
 *
 */
package indi.yangshenhui.hbase.test;

import java.io.BufferedReader;
import java.io.File;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

import org.apache.hadoop.hbase.client.HTableInterface;

import indi.yangshenhui.hbase.dao.HBaseBaseDao;
import indi.yangshenhui.hbase.dao.HBaseDaoImpl;
import indi.yangshenhui.hbase.dao.exception.HBaseDaoException;
import indi.yangshenhui.hbase.dao.pool.HTablePoolEngine;
import indi.yangshenhui.tool.DateUtil;
import indi.yangshenhui.tool.FileUtil;

/**
 * @ClassName RuJiaTask
 * @Description
 * @date 2016年4月9日 下午4:20:51
 * @author shenhui
 * @version 1.0.0
 *
 */
public class RuJiaTask implements Runnable {

	public void run() {
		RuJiaMain.jiaRnum();
		long start = System.currentTimeMillis();
		long start5000 = System.currentTimeMillis();

		System.err.println(Thread.currentThread().getName());
		HTableInterface hTable = null;
		File file = null;
		try {
			int c = 0;
			String line = "";
			String[] title = null;
			String[] content = null;
			List<RuJiaPojo> data = new ArrayList<RuJiaPojo>();

			while ((file = RuJiaMain.getFile()) != null) {
				String tName = RuJiaMain.cTable(file.getName());

				HBaseBaseDao hBaseDao = new HBaseDaoImpl();
				hTable = HTablePoolEngine.getHTable(tName);
				HTablePoolEngine.optimizeHTable(hTable, 5242880L);

				c = 0;

				System.err.println(file.getAbsolutePath());
				BufferedReader reader = FileUtil.getReader(file.getAbsolutePath(), "UTF-8");
				line = reader.readLine();
				++c;
				title = line.split(",");
				line = reader.readLine();
				++c;
				while (line != null) {
					content = line.split(",");
					if (title.length == content.length) {
						RuJiaPojo ruJiaPojo = new RuJiaPojo(content);
						data.add(ruJiaPojo);
					}

					if (data.size() >= (RuJiaMain.tc * 1000)) {
						hBaseDao.putObject(hTable, data);
						data.clear();
						System.err.println(
								Thread.currentThread().getName() + " " + DateUtil.format(new Date()) + " " + c + " put "
										+ (RuJiaMain.tc * 1000) + " cost:" + (System.currentTimeMillis() - start5000));
						start5000 = System.currentTimeMillis();
					}

					line = reader.readLine();
					++c;
				}

			}
		} catch (Exception e) {
			RuJiaMain.jianRnum();
			RuJiaMain.addFile(file);
			e.printStackTrace();
		} finally {
			try {
				HTablePoolEngine.closeHTable(hTable);
			} catch (HBaseDaoException e) {
				e.printStackTrace();
			}
		}
		long end = System.currentTimeMillis();
		System.err.println(">>> total cost:" + (end - start) + "ms");
	}

}
